#pragma once

#include "muduo/proto/dispatcher.h"
#include "muduo/proto/codec.h"
#include "muduo/base/Logging.h"
#include "muduo/base/Mutex.h"
#include "muduo/net/EventLoop.h"
#include "muduo/net/TcpClient.h"
#include "muduo/net/EventLoopThread.h"
#include "muduo/base/CountDownLatch.h"

#include "mq_channel.hpp"
#include "mq_worker.hpp"
#include "../mq_comm/mq_helper.hpp"

namespace mq
{
    using namespace ns_helper;
    class Connection
    {
    public:
        using ptr = std::shared_ptr<Connection>;
        Connection(const std::string &sip, int sport, const AsyncWorker::ptr &worker):
        _latch(1), _client(worker->loopthread.startLoop(), muduo::net::InetAddress(sip, sport), "Client"),
        _dispatcher(std::bind(&Connection::onUnknownMessage, this, std::placeholders::_1, 
            std::placeholders::_2, std::placeholders::_3)),
        _codec(std::make_shared<ProtobufCodec>(std::bind(&ProtobufDispatcher::onProtobufMessage, &_dispatcher, 
            std::placeholders::_1, std::placeholders::_2, std::placeholders::_3))),
        _worker(worker),
        _channel_manager(std::make_shared<ChannelManager>())
        {
            // 注册业务处理函数
            _dispatcher.registerMessageCallback<basicCommonResponse>(std::bind(&Connection::basicResponse, this,
                std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
            _dispatcher.registerMessageCallback<basicConsumeResponse>(std::bind(&Connection::consumeResponse, this,
                std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));

            // 设置client收到数据后的消息处理回调(这里是解析数据onMessage)
            _client.setMessageCallback(std::bind(&ProtobufCodec::onMessage, _codec.get(), 
                std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
            // 设置有新连接时的回调函数
            _client.setConnectionCallback(std::bind(&Connection::onConnection, this, std::placeholders::_1));

            _client.connect();
            _latch.wait();//阻塞等待，直到连接建立成功
        }
        
        Channel::ptr openChannel()
        {
            Channel::ptr channel = _channel_manager->create(_conn, _codec);
            // 客户端创建信道, 要与服务端信道对应, 所以这里需要通知服务端也创建与之对应的信道(发送打开信道的请求)
            bool ret = channel->openChannel();
            if (ret == false) 
            {
                LOG(ERROR) << "打开信道失败！\n";
                return Channel::ptr();
            }
            return channel;
        }
        void closeChannel(const Channel::ptr &channel) 
        {
            channel->closeChannel();
            _channel_manager->remove(channel->cid());
        }
    private:
        // 响应处理(发布者, 消费者)收到消息后自动调用
        void basicResponse(const muduo::net::TcpConnectionPtr& conn, const basicCommonResponsePtr& message, muduo::Timestamp) 
        {
            //1. 找到信道
            Channel::ptr channel = _channel_manager->get(message->cid());
            if (channel.get() == nullptr) 
            {
                LOG(ERROR) << "未找到信道信息！\n";
                return;
            }
            //2. 将得到的响应对象，添加到信道的基础响应hash_map中
            channel->putBasicResponse(message);
        }
        // 消费者业务处理(消息推送 -- 消费者) 自动将消息推送给消费者
        void consumeResponse(const muduo::net::TcpConnectionPtr& conn, const basicConsumeResponsePtr& message, muduo::Timestamp)
        {
            // 1.找到信道
            Channel::ptr channel = _channel_manager->get(message->cid());
            if (channel.get() == nullptr) 
            {
                LOG(ERROR) << "未找到信道信息！\n";
                return;
            }
            // 2.将任务进行封装, 进行异步处理(抛入线程池)
            _worker->pool.Push([channel, message](){
                channel->consume(message);
            });
        }
        void onUnknownMessage(const muduo::net::TcpConnectionPtr &conn,
                                const MessagePtr &message,
                                muduo::Timestamp)
        {
            LOG_INFO << "onUnknownMessage: " << message->GetTypeName();
            conn->shutdown();
        }

        void onConnection(const muduo::net::TcpConnectionPtr &conn)
        {
            if (conn->connected()) 
            {
                _conn = conn;
                _latch.countDown();//唤醒主线程中的阻塞
            }
            else 
            {
                //连接关闭时的操作
                _conn.reset();
            }
        }

    private:
        muduo::net::EventLoopThread _loop; // EventLoopThread对象在实例化完成之后直接就开始监控, 不需要我们自己去启动
        muduo::CountDownLatch _latch; // 内部使用条件变量进行同步控制
        muduo::net::TcpConnectionPtr _conn;
        muduo::net::TcpClient _client;
        ProtobufCodecPtr _codec; // 协议处理
        ProtobufDispatcher _dispatcher; // 任务派发

        AsyncWorker::ptr _worker;
        ChannelManager::ptr _channel_manager;
    };
}